package com.base.test.rabbitmq.service;

import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.base.commons.tools.utils.StringUtils;
import com.base.test.rabbitmq.constant.Constant;
import com.base.test.rabbitmq.entity.MqSendEntity;
import com.base.test.rabbitmq.properties.PublisherProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.UUID;

/**
 * 所有生产者父类
 * 封装消息重试机制
 * @author qiaodong
 * @since 2020-08-29
 */
@Slf4j
@EnableConfigurationProperties(PublisherProperties.class)
public abstract class AbstractProducerSender {

    @Autowired
    private MqSendService mqSendService;

    @Autowired
    private PublisherProperties properties;

    /**
     * 设置当前发送者
     * @return
     */
    public abstract AbstractProducerSender setSender();

    /**
     * 发送消息封装
     * @param exchange
     * @param routingKey
     * @param json
     * @param appCode
     * @param bussinessType
     * @return
     */
    @Transactional(rollbackFor = Exception.class ,propagation = Propagation.NOT_SUPPORTED)
    public boolean send(String msgId,String exchange,String routingKey,
                        String json,String appCode,String bussinessType){
            try{
                    //正常发送不需要msgId,重试还是第一次发送时的msgId
                    if(StringUtils.isEmpty(msgId)){
                        msgId = UUID.randomUUID().toString();
                    }
                    Message message = MessageBuilder.withBody(json.getBytes()).setCorrelationId(msgId).build();
                    //设置消息持久化
                    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                    //使用自定义的数据对象
                    CorrelationDataExt correlationData = new CorrelationDataExt();
                    correlationData.setId(msgId);
                    correlationData.setMsgJson(json);
                    correlationData.setExchange(exchange);
                    correlationData.setRoutingKey(routingKey);
                    correlationData.setAppCode(appCode);
                    correlationData.setBussinessType(bussinessType);
                    MqSendEntity msg =  makeMsg(msgId,exchange,routingKey,
                            json,appCode,bussinessType, Constant.ONE,Constant.ZERO,
                            Constant.ZERO,"消息发送之前入库");
                    //重试场景先查看DB是否已经保存了已发送消息
                    int result = Constant.ZERO;
                    MqSendEntity mqSend = mqSendService.getByMsgId(msgId);
                    if(null==mqSend){
                        result = mqSendService.checkSendInsertOrUpdate(msg);
                    }else{
                        //乐观锁进行重试次数累加
                        result = mqSendService.addRetryCount(msgId,mqSend.getRetryCount(),"发送消息重试");
                    }
                    if(result > Constant.ZERO){//消息表插入成功后再发送消息
                        realitySend(exchange,routingKey,message,correlationData);
                    }
            }catch(Exception e){
                log.error("MQ连接失败，请联系管理员处理!!!!");
                MqSendEntity msg =  makeMsg(msgId,exchange,routingKey,
                        json,appCode,bussinessType,Constant.TWO,Constant.ONE,
                        Constant.ZERO,"MQ连接失败");
                mqSendService.checkSendInsertOrUpdate(msg);
                return false;
            }
        return true;
    }

    /**
     * 实际发送消息方法
     * @param exchange
     * @param routingKey
     * @param message
     * @param correlationData
     */
    public abstract void realitySend(String exchange,String routingKey,
                                     Message message,CorrelationDataExt correlationData);


    /**
     * 发布者确认
     * @param correlationData
     * @param ack
     * @param cause
     */
    public void confirm(CorrelationData correlationData, boolean ack, String cause){
        String json = "";
        String exchange ="";
        String routingKey = "";
        String appCode = "";
        String bussinessType = "";
        String msgId = correlationData.getId();
        if(correlationData instanceof CorrelationDataExt){
            CorrelationDataExt correlationDataExt = (CorrelationDataExt) correlationData;
            json = (String) correlationDataExt.getMsgJson();
            exchange = ((CorrelationDataExt) correlationData).getExchange();
            routingKey = ((CorrelationDataExt) correlationData).getRoutingKey();
            appCode = ((CorrelationDataExt) correlationData).getAppCode();
            bussinessType = ((CorrelationDataExt) correlationData).getBussinessType();
        }
        if(ack){
            //发送成功
            log.debug("ack,消息投递到exchange成功,msgId:{}",msgId);
            MqSendEntity msg =  makeMsg(msgId,exchange,routingKey,
                    json,appCode,bussinessType, Constant.THREE,Constant.ZERO,
                    Constant.ZERO,"消息投递到exchange成功");
            mqSendService.checkSendInsertOrUpdate(msg);
        }else{
            log.error("ack,消息投递exchange失败，msgId:{},原因{}" ,msgId, cause);
            //开启发布者重试，才进行发送失败重试
            if(properties.isRetryEnable()){
                //获取重试次数配置
                Integer retryCount = properties.getRetryCount();
                try{
                    MqSendEntity mqSend = mqSendService.getByMsgId(msgId);
                    if(mqSend.getRetryCount() < retryCount){
                        setSender().send(msgId,exchange,routingKey,json,appCode,bussinessType);
                    }
                }catch (Exception e){
                    log.error("生产者消息重试失败");
                }
            }
        }
    }

    /**
     * 消息投递队列回调
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    public void returnedMessage(Message message,int replyCode,String replyText,
                                String exchange,String routingKey){
        log.error("消息发送失败-消息回退，应答码：{}，原因：{}，交换机：{}，路由键：{}",
                replyCode, replyText, exchange, routingKey);
        String msgId = message.getMessageProperties().getCorrelationId();
        String json = new String(message.getBody());
        //投递队列失败
        MqSendEntity msg =  makeMsg(msgId,exchange,routingKey,
                json,null,null,Constant.TWO,Constant.THREE,
                Constant.ZERO,replyText);
        mqSendService.checkSendInsertOrUpdate(msg);
    }

    /**
     * 构建消息内容
     * @param msgId
     * @param exchange
     * @param routingKey
     * @param json
     * @param appCode
     * @param bussinessType
     * @param status
     * @param retryCount
     * @param remark
     * @return
     */
    private MqSendEntity makeMsg(String msgId, String exchange, String routingKey,
                                String json, String appCode, String bussinessType,
                                Integer status, Integer failType, Integer retryCount,
                                String remark){
        MqSendEntity mqSend = new MqSendEntity();
        mqSend.setId(IdWorker.getId());
        mqSend.setMsgId(msgId);
        mqSend.setExchange(exchange);
        mqSend.setRoutingKey(routingKey);
        mqSend.setMsgJson(json);
        mqSend.setAppCode(appCode);
        mqSend.setBusinessType(bussinessType);
        mqSend.setStatus(status);
        mqSend.setFailType(failType);
        mqSend.setRetryCount(retryCount);
        mqSend.setRemark(remark);
        return mqSend;
    }
}
